package com.kxn.multi.module.service.mq.consumer;

import com.kxn.multi.module.common.constant.MqConstant;
import com.kxn.multi.module.service.dto.api.mq.CommonMessageParam;
import com.kxn.multi.module.service.mq.consumer.handler.MqBaseHandler;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * 通用的消息处理
 *
 * 注意：ConsumeMode.ORDERLY 顺序模式，如果消费者数*线程数大于队列数，会导致超出的部分闲置
 *
 */
@Slf4j
@Component
@RocketMQMessageListener(topic = MqConstant.COMMON_TOPIC, consumerGroup = MqConstant.COMMON_TOPIC_CONSUMER_GROUP,
        consumeMode = ConsumeMode.ORDERLY, consumeThreadNumber = 2)
public class CommonConsumer implements RocketMQListener<CommonMessageParam>, RocketMQPushConsumerLifecycleListener {


    @Value("${rocketmq.max-reconsume-times:3}")
    private Integer maxReconsumeTimes;

    @Autowired
    private Map<String, MqBaseHandler> mqHandlerMap;


    @Override
    public void prepareStart(DefaultMQPushConsumer consumer) {
        // 设置消费最大重试次数
        consumer.setMaxReconsumeTimes(maxReconsumeTimes);
    }

    @Override
    public void onMessage(CommonMessageParam commonMessageParam) {
        String message = commonMessageParam.getMessage();
        String handler = commonMessageParam.getHandler();
        log.info("topic:{}, 获取到的消息：{}, handler:{}", MqConstant.COMMON_TOPIC, message, handler);

        if (StringUtils.isBlank(handler)) {
            log.error("handler错误：topic:{}, handler:{}, message：{},", MqConstant.COMMON_TOPIC,commonMessageParam.getHandler(), message);
            return;
        }

        if (!mqHandlerMap.containsKey(handler)) {
            log.error("handler不存在：topic:{}, handler:{}, message：{},", MqConstant.COMMON_TOPIC,commonMessageParam.getHandler(), message);
            return;
        }

        MqBaseHandler mqHandler = mqHandlerMap.get(handler);
        mqHandler.run();
    }

}
